package com.verlet.springbootdemo.simple.disruptor.netty;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

public class RingBufferWorkPoolFactory {

    private static volatile RingBufferWorkPoolFactory instance;

    private RingBufferWorkPoolFactory() {

    }

    public static RingBufferWorkPoolFactory getInstance() {
        if (instance == null) {
            synchronized (RingBufferWorkPoolFactory.class) {
                if (instance == null) {
                    instance = new RingBufferWorkPoolFactory();
                }
            }
        }
        return instance;
    }

    private static Map<String, MessageProducer> producers = new ConcurrentHashMap<>();

    private static Map<String, AbstractMessageConsumer> consumers = new ConcurrentHashMap<>();

    private RingBuffer<TranslatorDataWapper> ringBuffer;

    private SequenceBarrier sequenceBarrier;

    private WorkerPool<TranslatorDataWapper> workerPool;


    public void initAndStart(ProducerType type, int bufferSize, WaitStrategy waitStrategy, AbstractMessageConsumer[] messageConsumers) {
        ringBuffer = RingBuffer.create(type, new EventFactory<TranslatorDataWapper>() {
            @Override
            public TranslatorDataWapper newInstance() {
                return new TranslatorDataWapper();
            }
        }, bufferSize, waitStrategy);

        sequenceBarrier = ringBuffer.newBarrier();

        workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, new IntEventExceptionHandler(), messageConsumers);

        for (AbstractMessageConsumer messageConsumer : messageConsumers) {
            consumers.put(messageConsumer.getConsumerId(), messageConsumer);
        }

        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));
    }

    public MessageProducer getMessageProducer(String producerId) {
        MessageProducer messageProducer = producers.get(producerId);
        if (messageProducer == null) {
            messageProducer = new MessageProducer(producerId, ringBuffer);
            producers.put(producerId, messageProducer);
        }

        return messageProducer;
    }


    private static class IntEventExceptionHandler implements ExceptionHandler<TranslatorDataWapper> {

        // 进行消费的时候出现异常
        @Override
        public void handleEventException(Throwable ex, long sequence, TranslatorDataWapper event) {

        }

        // 启动时候
        @Override
        public void handleOnStartException(Throwable ex) {

        }

        // 消费结束的时候
        @Override
        public void handleOnShutdownException(Throwable ex) {

        }
    }

}
